package beautiful.butterfly.drds.data_exchange.reader.util;

import beautiful.butterfly.drds.data_exchange.constant.CommonConstant;
import beautiful.butterfly.drds.data_exchange.reader.Constant;
import beautiful.butterfly.drds.data_exchange.reader.Key;
import beautiful.butterfly.drds.data_exchange.util.Configuration;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

public final class ReaderSplitUtil
{


    public static List<Configuration> doSplit(
            Configuration originalSliceConfig, int adviceNumber) throws SQLException
    {
        boolean isTableMode = originalSliceConfig.getBoolean(Constant.IS_TABLE_MODE).booleanValue();
        int eachTableShouldSplittedNumber = -1;
        if (isTableMode)
        {
            // adviceNumber这里是channel数量大小, 即datax并发task数量
            // eachTableShouldSplittedNumber是单表应该切分的份数, 向上取整可能和adviceNumber没有比例关系了已经
            eachTableShouldSplittedNumber = calculateEachTableShouldSplittedNumber(
                    adviceNumber, originalSliceConfig.getInt(Constant.TABLE_NUMBER_MARK));
        }

        String column = originalSliceConfig.getString(Key.column);
        String where = originalSliceConfig.getString(Key.where, null);

        List<Object> conns = originalSliceConfig.getList(Constant.CONN_MARK, Object.class);

        List<Configuration> splittedConfigs = new ArrayList<Configuration>();

        for (int i = 0, len = conns.size(); i < len; i++)
        {
            Configuration sliceConfig = originalSliceConfig.clone();

            Configuration configuration = Configuration.from(conns.get(i).toString());
            String jdbcUrl = configuration.getString(Key.jdbc_url);
            sliceConfig.set(Key.jdbc_url, jdbcUrl);

            // 抽取 jdbcUrl 中的 ip/port 进行资源使用的打标，以提供给 core 做有意义的 shuffle 操作
            sliceConfig.set(CommonConstant.load_balance_resource_mark, jdbcUrl);//ip

            sliceConfig.remove(Constant.CONN_MARK);

            Configuration tempSlice;

            // 说明是配置的 tableName 方式
            if (isTableMode)
            {
                // 已在之前进行了扩展和`处理，可以直接使用
                List<String> tables = configuration.getList(Key.table_name, String.class);

                Validate.isTrue(null != tables && !tables.isEmpty(), "您读取数据库表配置错误.");

                String splitPk = originalSliceConfig.getString(Key.split_key, null);

                //最终切分份数不一定等于 eachTableShouldSplittedNumber
                boolean needSplitTable = eachTableShouldSplittedNumber > 1
                        && StringUtils.isNotBlank(splitPk);
                if (needSplitTable)
                {
                    if (tables.size() == 1)
                    {
                        //原来:如果是单表的，主键切分num=num*2+1
                        // splitPk is null这类的情况的数据量本身就比真实数据量少很多, 和channel大小比率关系时，不建议考虑
                        //eachTableShouldSplittedNumber = eachTableShouldSplittedNumber * 2 + 1;// 不应该加1导致长尾

                        //考虑其他比率数字?(splitPk is null, 忽略此长尾)
                        eachTableShouldSplittedNumber = eachTableShouldSplittedNumber * 5;
                    }
                    // 尝试对每个表，切分为eachTableShouldSplittedNumber 份
                    for (String table : tables)
                    {
                        tempSlice = sliceConfig.clone();
                        tempSlice.set(Key.table_name, table);

                        List<Configuration> splittedSlices = SingleTableSplitUtil
                                .splitSingleTable(tempSlice, eachTableShouldSplittedNumber);

                        splittedConfigs.addAll(splittedSlices);
                    }
                } else
                {
                    for (String table : tables)
                    {
                        tempSlice = sliceConfig.clone();
                        tempSlice.set(Key.table_name, table);
                        String queryColumn = column;
                        tempSlice.set(Key.query_sql, SingleTableSplitUtil.buildQuerySql(queryColumn, table, where));
                        splittedConfigs.add(tempSlice);
                    }
                }
            } else
            {
                // 说明是配置的 querySql 方式
                List<String> sqls = configuration.getList(Key.query_sql, String.class);

                // TODO 是否check 配置为多条语句？？
                for (String querySql : sqls)
                {
                    tempSlice = sliceConfig.clone();
                    tempSlice.set(Key.query_sql, querySql);
                    splittedConfigs.add(tempSlice);
                }
            }

        }

        return splittedConfigs;
    }

    public static Configuration doPreCheckSplit(Configuration originalSliceConfig)
    {
        Configuration queryConfig = originalSliceConfig.clone();
        boolean isTableMode = originalSliceConfig.getBoolean(Constant.IS_TABLE_MODE).booleanValue();

        String splitPK = originalSliceConfig.getString(Key.split_key);
        String column = originalSliceConfig.getString(Key.column);
        String where = originalSliceConfig.getString(Key.where, null);

        List<Object> conns = queryConfig.getList(Constant.CONN_MARK, Object.class);

        for (int i = 0, len = conns.size(); i < len; i++)
        {
            Configuration connConf = Configuration.from(conns.get(i).toString());
            List<String> querys = new ArrayList<String>();
            List<String> splitPkQuerys = new ArrayList<String>();
            String connPath = String.format("connection[%d]", i);
            // 说明是配置的 tableName 方式
            if (isTableMode)
            {
                // 已在之前进行了扩展和`处理，可以直接使用
                List<String> tables = connConf.getList(Key.table_name, String.class);
                Validate.isTrue(null != tables && !tables.isEmpty(), "您读取数据库表配置错误.");
                for (String table : tables)
                {
                    querys.add(SingleTableSplitUtil.buildQuerySql(column, table, where));
                    if (splitPK != null && splitPK.length() > 0)
                    {
                        splitPkQuerys.add(SingleTableSplitUtil.genPKSql(splitPK.trim(), table, where));
                    }
                }
                if (!splitPkQuerys.isEmpty())
                {
                    connConf.set(Key.split_key_sql, splitPkQuerys);
                }
                connConf.set(Key.query_sql, querys);
                queryConfig.set(connPath, connConf);
            } else
            {
                // 说明是配置的 querySql 方式
                List<String> sqls = connConf.getList(Key.query_sql,
                        String.class);
                for (String querySql : sqls)
                {
                    querys.add(querySql);
                }
                connConf.set(Key.query_sql, querys);
                queryConfig.set(connPath, connConf);
            }
        }
        return queryConfig;
    }

    private static int calculateEachTableShouldSplittedNumber(int adviceNumber,
                                                              int tableNumber)
    {
        double tempNum = 1.0 * adviceNumber / tableNumber;

        return (int) Math.ceil(tempNum);
    }

}
